package com.lvw.weather.city.service.impl;

import com.lvw.weather.city.service.ICityService;
import com.netflix.hystrix.*;
import com.netflix.hystrix.strategy.concurrency.HystrixConcurrencyStrategyDefault;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Service;
import org.springframework.web.client.RestTemplate;
import rx.Observable;
import rx.Subscriber;

import java.util.Collection;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.logging.Logger;
import java.util.stream.Collectors;

/**
 * Hystrix的编程实现版本
 */
@Service
@Qualifier("cityServiceHystrixRough")
public class CityServiceHystrixRough implements ICityService {

    final Logger log = Logger.getLogger(getClass().toString());

    @Autowired
    RestTemplate restTemplate;

    @Override
    public String cityForcast(String city) {

        String forcast = null;

        com.netflix.hystrix.HystrixCommand.Setter setter = com.netflix.hystrix.HystrixCommand.Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("GroupName"))
                .andCommandKey(HystrixCommandKey.Factory.asKey("forcastCommand"));

        // sync invoke.
        //forcast = new ForcastCommand(setter, restTemplate, city).execute();

        /*
        // async invoke.
        Future<String> futureRet = new ForcastCommand(setter, restTemplate, city).queue();
        try {
            forcast = futureRet.get();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }*/

        /*
        // rxjava mode
        // hot observable, 立即执行
        Observable<String> hotObservable = new ForcastCommand(setter, restTemplate, city).observe();
        // cold observable, 有了订阅在执行
        Observable<String> coldObservable = new ForcastCommand(setter, restTemplate, city).toObservable();*/

        // 测试批量处理
        Future<String> s1 = new ForcastCollapserCommand(restTemplate, "Shen").queue();
        Future<String> s2 = new ForcastCollapserCommand(restTemplate, "Hei").queue();
        Future<String> s3 = new ForcastCollapserCommand(restTemplate, "Qing").queue();

        try {
            forcast = s1.get();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }


        return forcast;
    }

}

// 命令模式
class ForcastCommand extends HystrixCommand<String> {

    final Logger log = Logger.getLogger(getClass().toString());

    private static final HystrixCommandKey GETTER_KEY = HystrixCommandKey.Factory.asKey("forcastCommand");

    private RestTemplate restTemplate;
    private String city;

    protected ForcastCommand(Setter setter, RestTemplate restTemplate, String city) {
        super(setter);
        this.restTemplate = restTemplate;
        this.city = city;
    }

    @Override
    protected String run() throws Exception {
        return restTemplate.getForEntity("http://WEATHER-FORCAST-SERVICE/forcast/" + city, String.class).getBody();
    }

    @Override
    protected String getFallback() {
        log.info("FORCAST SERVICE FALLBACK");
        return "error";
    }

    // enabled cache
    @Override
    protected String getCacheKey() {
        return city;
    }

    // 对于写操作, 可以刷新缓存
    public static void flushCache(String city) {
        HystrixRequestCache.getInstance(GETTER_KEY, HystrixConcurrencyStrategyDefault.getInstance())
                .clear(city);
    }
}

// 批量请求的命令模式
class ForcastBatchCommand extends HystrixCommand<List<String>> {

    final Logger log = Logger.getLogger(getClass().toString());

    private RestTemplate restTemplate;
    private List<String> citys;

    protected ForcastBatchCommand(RestTemplate restTemplate, List<String> citys) {
        super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("forcastBatchCommand")));
        this.restTemplate = restTemplate;
        this.citys = citys;
    }

    @Override
    protected List<String> run() throws Exception {
        return restTemplate.getForEntity("http://WEATHER-FORCAST-SERVICE/forcast?citys=" + citys, List.class).getBody();
    }

    @Override
    protected List<String> getFallback() {
        log.info("FORCAST SERVICE FALLBACK");

        return citys.stream().map(s -> "error").collect(Collectors.toList());
    }

}

/**
 * 批量处理
 */
class ForcastCollapserCommand extends HystrixCollapser<List<String>, String, String> {

    final Logger log = Logger.getLogger(getClass().toString());


    private RestTemplate restTemplate;
    private String city;

    public ForcastCollapserCommand(RestTemplate restTemplate, String city) {

        // 设定100ms内的请求可以合并转发
        super(Setter.withCollapserKey(HystrixCollapserKey.Factory.asKey("forcastCollapserCommand"))
                .andCollapserPropertiesDefaults(HystrixCollapserProperties.Setter().withTimerDelayInMilliseconds(500)));
        this.restTemplate = restTemplate;
        this.city = city;
    }

    @Override
    public String getRequestArgument() {
        return city;
    }

    @Override
    protected HystrixCommand<List<String>> createCommand(Collection<CollapsedRequest<String, String>> collapsedRequests) {
        System.out.println("CollapserCommand========>");
        List<String> list = collapsedRequests.stream().map(CollapsedRequest::getArgument).collect(Collectors.toList());
        return new ForcastBatchCommand(restTemplate, list);
    }

    @Override
    protected void mapResponseToRequests(List<String> batchResponse, Collection<CollapsedRequest<String, String>> collapsedRequests) {

        // batchResponse 依赖服务的返回值, 这个值需要设置到 collapsedRequests中, collapsedRequests会返回给各个调用的客户端\
        int i = 0;
        for (CollapsedRequest<String, String> collapsedRequest : collapsedRequests) {
            collapsedRequest.setResponse(batchResponse.get(i++));
        }

    }
}

// update db
class ForcastPostCommand extends HystrixCommand<String> {

    final Logger log = Logger.getLogger(getClass().toString());

    private static final HystrixCommandKey GETTER_KEY = HystrixCommandKey.Factory.asKey("forcastPostCommand");

    private RestTemplate restTemplate;
    private String city;

    protected ForcastPostCommand(Setter setter, RestTemplate restTemplate, String city) {
        super(setter);
        this.restTemplate = restTemplate;
        this.city = city;
    }

    @Override
    protected String run() throws Exception {
        // update db omitted

        // flush cache
        ForcastCommand.flushCache(city);
        return "";
    }

    @Override
    protected String getFallback() {
        log.info("FORCAST SERVICE FALLBACK");
        return "error";
    }

}


// rxjava
class ForcastCommandObservable extends HystrixObservableCommand<String> {

    final Logger log = Logger.getLogger(getClass().toString());
    private RestTemplate restTemplate;
    private String city;

    protected ForcastCommandObservable(Setter setter, RestTemplate restTemplate, String city) {
        super(setter);
        this.restTemplate = restTemplate;
        this.city = city;
    }


    @Override
    protected Observable<String> resumeWithFallback() {
        return Observable.create(new Observable.OnSubscribe<String>() {
            @Override
            public void call(Subscriber<? super String> subscriber) {
                subscriber.onNext("error");
                subscriber.onCompleted();
            }
        });
    }

    @Override
    protected Observable<String> construct() {
        return Observable.create(new Observable.OnSubscribe<String>() {

            @Override
            public void call(Subscriber<? super String> subscriber) {

                if (!subscriber.isUnsubscribed()) {
                    String string = restTemplate.getForEntity("http://WEATHER-FORCAST-SERVICE/forcast/" + city, String.class).getBody();
                    subscriber.onNext(string);
                    subscriber.onCompleted();
                }

            }
        });
    }
}